package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcess;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.io.IOException;

/**
 * @Author atguigu
 * @Date 2023/7/4 14:00
 */
public class HBaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private Jedis jedis;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }
    
    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConn(conn);
        RedisUtil.closeJedis(jedis);
        
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t,
                       Context ctx) throws Exception {
        // insert update delete
        TableProcess tp = t.f1;
        JSONObject data = t.f0;
        String opType = data.getString("op_type");
    
        if ("delete".equals(opType)) {
            // 删除维度信息
            delDim(t);
        }else{
            // insert 和 update 的时候, 写入维度数据
            putDim(t);
        }
        
        
        // 如果是维度是 update 或 delete 则删除缓存中的维度数据
        if ("delete".equals(opType) || "update".equals(opType)) {
            String key = RedisUtil.getKey(tp.getSinkTable(), data.getString("id"));
            jedis.del(key);
        }
    }
    
    private void putDim(Tuple2<JSONObject, TableProcess> t) throws IOException {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        
        String rowKey = data.getString(tp.getSinkRowKey());
        // data中有多少 kv 就写多少列 - 1
        JSONObject dataCopy = new JSONObject();
        dataCopy.putAll(data);
        dataCopy.remove("op_type");
        System.out.println(t);
        HBaseUtil.putRow(conn,
                         Constant.HBASE_NAMESPACE,
                         tp.getSinkTable(),
                         rowKey,
                         tp.getSinkFamily(),
                         dataCopy);
        
    }
    
    private void delDim(Tuple2<JSONObject, TableProcess> t) throws IOException {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
    
        String rowKey = data.getString(tp.getSinkRowKey());
        
        HBaseUtil.delRow(conn,
                         Constant.HBASE_NAMESPACE,
                         tp.getSinkTable(),
                         rowKey);
    }
}
